﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Net.Sockets;
using System.IO;
using System.Threading;
using Platform.Storage.Redis.Messages;

namespace Platform.Storage.Redis
{

	/// <summary>
	/// Objeto utilizado para enviar e receber as mensagens do REDIS
	/// </summary>
	public sealed class RedisMessageChannel
	{
		private const string EndOfArgument = "\r\n";

		private Socket connection;
		private TextWriter output;
		private TextReader input;
		private Queue<IRedisMessage> waitingQueue = new Queue<IRedisMessage>();

		internal RedisMessageChannel()
		{
			this.Stats = new ChannelStats();
		}

		/// <summary>
		/// Contadores do canal
		/// </summary>
		public ChannelStats Stats
		{ get; private set; }

		/// <summary>
		/// Indica se está conectado
		/// </summary>
		public bool Connected
		{
			get
			{
				return this.connection != null
					&& this.connection.Connected;
			}
		}

		/// <summary>
		/// Indica se a conexão possui dados para ser processado
		/// </summary>
		public bool DataAvailable
		{
			get
			{
				return this.Connected
					&& this.connection.Available > 0;
			}
		}

		public int ExecuteInt32( string command, params string[] parameters )
		{
			return this.Send( new Int32RedisMessage( this, command, parameters ) ).Reply;
		}

		public long ExecuteInt64( string command, params string[] parameters )
		{
			return this.Send( new Int64RedisMessage( this, command, parameters ) ).Reply;
		}

		public string ExecuteSingle( string command, params string[] parameters )
		{
			return this.Send( new StringRedisMessage( this, command, parameters ) ).Reply;
		}

		public IEnumerable<string> ExecuteOrQuery( string command, params string[] parameters )
		{
			return this.Send( new CollectionRedisMessage( this, command, parameters ) ).Reply;
		}

		/// <summary>
		/// Envia a mensagem para o REDIS
		/// </summary>
		/// <param name="message"></param>
		public TMessage Send<TMessage>( TMessage message )
			where TMessage : IRedisMessage
		{
			string data = message.ToString();

			if ( !String.IsNullOrEmpty( data ) )
			{
				lock ( this.output )
				{
					this.output.Write( data );
					this.waitingQueue.Enqueue( message );
				}
				this.output.Flush();

				this.Stats.MessageSent( data.Length );
			}
			return message;
		}

		/// <summary>
		/// Envia a mensagem e aguarda a resposta
		/// </summary>
		/// <typeparam name="TMessage"></typeparam>
		/// <param name="message"></param>
		/// <returns></returns>
		public TMessage SendAndWaitReply<TMessage>( TMessage message )
			where TMessage : IRedisMessage
		{
			this.Send<TMessage>( message );
			this.WaitReplyOf( message );
			return message;
		}

		//public TMessage SendAndWaitReply<TMessage>( RedisMessageBase<TMessage message )
		//	where TMessage : IRedisMessage
		//{
		//	this.Send
		//}

		/// <summary>
		/// Espera chegar todas as mensagens enviadas
		/// </summary>
		public void WaitAllReplies()
		{
			IRedisMessage last = null;

			lock ( this.input )
				last = this.waitingQueue.LastOrDefault();

			if ( last != null )
				this.WaitReplyOf( last );
		}

		/// <summary>
		/// Aguarda até receber a resposta do REDIS
		/// </summary>
		/// <param name="message"></param>
		public void WaitReplyOf( IRedisMessage message )
		{
			while ( this.IsQueued( message ) )
			{
				IRedisMessage nextMessage = null;

				lock ( this.input )
				{
					// A mensagem continua aguardando para ser processada?
					if ( this.IsQueued( message ) )
					{
						nextMessage = this.waitingQueue.Dequeue();
						this.ReadReplyOf( nextMessage );
					}
				}

				// A mensagem já foi processada por outra thread
				// ou chegou na mensagem que estava aguardando para ser processada
				if ( nextMessage == null || nextMessage.Equals( message ) )
					return;
			}
		}

		/// <summary>
		/// 
		/// </summary>
		/// <param name="message"></param>
		private void ReadReplyOf( IRedisMessage message )
		{
			lock ( this.input )
				message.ReadReply( this.input );
		}

		/// <summary>
		/// 
		/// </summary>
		/// <param name="message"></param>
		/// <returns></returns>
		public bool WaitNotification( out ChannelNotificationMessage message )
		{
			return this.WaitNotification( TimeSpan.Zero, out message );
		}

		/// <summary>
		/// 
		/// </summary>
		/// <param name="timeout"></param>
		/// <param name="message"></param>
		/// <returns></returns>
		public bool WaitNotification( TimeSpan timeout, out ChannelNotificationMessage message )
		{
			message = null;

			if ( timeout > TimeSpan.Zero )
			{
				Thread.Sleep( timeout );
				if ( !this.DataAvailable )
					return false;
			}
			else
				while ( !this.DataAvailable )
					Thread.Sleep( 100 );

			message = new ChannelNotificationMessage( this );
			this.ReadReplyOf( message );
			return true;
		}

		/// <summary>
		/// Verifica se a mensagem está aguardando para ser processada
		/// </summary>
		/// <param name="message"></param>
		/// <returns></returns>
		public bool IsQueued( IRedisMessage message )
		{
			return this.waitingQueue.Contains( message );
		}

		/// <summary>
		/// 
		/// </summary>
		/// <param name="connection"></param>
		internal void ResetConnection( Socket connection )
		{
			this.connection = connection;

			this.connection = connection;
			this.output = new StreamWriter( new BufferedStream( new NetworkStream( connection, false ), 1024 ), Encoding.Default );
			this.input = new StreamReader( new NetworkStream( connection, false ), Encoding.Default );
		}

		/// <summary>
		/// 
		/// </summary>
		internal void CloseStreams()
		{
			this.connection = null;

			if ( this.output != null )
			{
				this.output.Dispose();
				this.output = null;
			}

			if ( this.input != null )
			{
				this.input.Dispose();
				this.input = null;
			}
		}
	}

	
}
